/********************************************************/
/*	Author:		gong_libin			*/
/*	Date:		2017_11_01			*/
/*	File:		CmKafka.h			*/
/********************************************************/

#ifndef	_CMKAFKA_H
#define	_CMKAFKA_H

#include <string>
#include "CmQueue.h"
extern "C" {
#include <librdkafka/rdkafka.h>
}

#define CM_KFK_TOUT			2000

#define	CM_KFK_PROD			RD_KAFKA_PRODUCER
#define	CM_KFK_CONS			RD_KAFKA_CONSUMER
#define	CM_KFK_PART			RD_KAFKA_PARTITION_UA

#define	CM_KFK_GZIP			(char*)"gzip"
#define	CM_KFK_BRKR			(char*)"broker"
#define	CM_KFK_SNAP			(char*)"snappy"
#define	CM_KFK_GPID			(char*)"group.id"
#define	CM_KFK_COMP			(char*)"compression.codec"
#define	CM_KFK_OFFS			(char*)"offset.store.method"
#define	CM_KFK_ERRS			(char*)"Failed to CmKafkaInit"

typedef std::string			CMSTRING_C;

class CCmKafka
{
public:
	CCmKafka() {NULL != (m_pstCfg = CmKafkaConfNew()) && NULL != (m_pstTfg = CmKafkaTopicConfNew()) ? NULL : throw(CM_KFK_ERRS);}
	virtual ~CCmKafka() {CmKafkaTopicDestroy();CmKafkaDestroy();}

	void CmKafkaDestroy() {rd_kafka_destroy(m_pstKfk);}
	void CmKafkaPoll(int iTime) {rd_kafka_poll(m_pstKfk, iTime);}
	void CmKafkaTopicDestroy() {rd_kafka_topic_destroy(m_pstTpc);}
	rd_kafka_conf_t* CmKafkaConfNew() {return rd_kafka_conf_new();}
	int CmKafkaDestroyWait(int iTime) {return rd_kafka_wait_destroyed(iTime);}
	int CmKafkaBroker(char* pszBrk) {return rd_kafka_brokers_add(m_pstKfk, pszBrk);}
	rd_kafka_topic_conf_t* CmKafkaTopicConfNew() {return rd_kafka_topic_conf_new();}
	rd_kafka_t* CmKafkaNew(rd_kafka_type_t iType) {return rd_kafka_new(iType, m_pstCfg, NULL, 0);}
	rd_kafka_topic_t* CmKafkaTopicNew(char* pszTpc) {return rd_kafka_topic_new(m_pstKfk, pszTpc, m_pstTfg);}
	bool CmKafkaTopicConfSetDefault() {rd_kafka_conf_set_default_topic_conf(m_pstCfg, m_pstTfg); return true;}
	int CmKafkaConfSet(char* pszKey, char* pszVal) {return rd_kafka_conf_set(m_pstCfg, pszKey, pszVal, NULL, 0);}
	int CmKafkaTopicConfSet(char* pszKey, char* pszVal) {return rd_kafka_topic_conf_set(m_pstTfg, pszKey, pszVal, NULL, 0);}

protected:
	rd_kafka_t*		m_pstKfk;
	rd_kafka_conf_t*	m_pstCfg;
	rd_kafka_topic_t*	m_pstTpc;
	rd_kafka_topic_conf_t*	m_pstTfg;

	virtual bool CmKafkaInit(char* pszBrk, char* pszTpc) {return NULL != (m_pstTpc = CmKafkaTopicNew(pszTpc)) && 0 != CmKafkaBroker(pszBrk);}

private:

};

class CCmProducer : public CCmKafka
{
public:
	CCmProducer(char* pszBrk, char* pszTpc) : CCmKafka() {true == CmKafkaInit(pszBrk, pszTpc) ? NULL : throw(CM_KFK_ERRS);}
	virtual ~CCmProducer() {}

	void CmProducerSend(UINT uiPart, char* pszMsg, int iSize) {rd_kafka_produce(m_pstTpc, uiPart, RD_KAFKA_MSG_F_COPY, pszMsg, iSize, NULL, 0, NULL);}

protected:
	virtual bool CmKafkaInit(char* pszBrk, char* pszTpc) {return NULL != (m_pstKfk = CmKafkaNew(CM_KFK_PROD)) && CCmKafka::CmKafkaInit(pszBrk, pszTpc);}

private:

};

class CCmConsumer : public CCmKafka
{
public:
	CCmConsumer(char* pszBrk, char* pszTpc, char* pszGid) : CCmKafka() {
		(RD_KAFKA_CONF_OK == CmKafkaConfSet(CM_KFK_GPID, pszGid) ? true : false) &&
		(RD_KAFKA_CONF_OK == CmKafkaTopicConfSet(CM_KFK_OFFS, CM_KFK_BRKR) ? true : false) &&
		true == CmKafkaTopicConfSetDefault() &&
		true == CmKafkaInit(pszBrk, pszTpc) ? NULL : throw(CM_KFK_ERRS);
	}
	virtual ~CCmConsumer() {rd_kafka_topic_partition_list_destroy(m_pstList);}

	CmCls::CCmQueue* CmConsumerRecv() {
		CmCls::CCmQueue* pCQueue = &m_CQueue;
		rd_kafka_message_t* pstMsg = rd_kafka_consumer_poll(m_pstKfk, CM_KFK_TOUT);

		if (NULL != pstMsg) {
			if (pstMsg->len > 0) {
				CMSTRING_C* pCMsg = new CMSTRING_C((char*)pstMsg->payload, pstMsg->len);
				m_CQueue.CmQueuePut(&pCMsg, sizeof(CMSTRING_C**));
			}
			else {
				pCQueue = NULL;
			}
			rd_kafka_message_destroy(pstMsg);
		}
		else {
			pCQueue = NULL;
		}

		return pCQueue;
	}

protected:
	virtual bool CmKafkaInit(char* pszBrk, char* pszTpc) {
		m_pstKfk = CmKafkaNew(CM_KFK_CONS);
		rd_kafka_poll_set_consumer(m_pstKfk);
		m_pstList = rd_kafka_topic_partition_list_new(1);
		rd_kafka_topic_partition_list_add(m_pstList, pszTpc, CM_KFK_PART);
		rd_kafka_subscribe(m_pstKfk, m_pstList);

		return CCmKafka::CmKafkaInit(pszBrk, pszTpc);
	}

private:
	CmCls::CCmQueue				m_CQueue;
	rd_kafka_topic_partition_list_t*	m_pstList;

};

#endif /* _CMKAFKA_H */
